package org.apache.linkis.engineplugin.spark.utils

import org.apache.hadoop.fs.Path
import org.apache.hadoop.fs.permission.{FsAction, FsPermission}
import org.apache.linkis.common.conf.Configuration.GATEWAY_URL
import org.apache.linkis.hadoop.common.utils.HDFSUtils
import org.slf4j.LoggerFactory
import org.springframework.http.{HttpEntity, HttpHeaders, HttpMethod, MediaType}
import org.springframework.web.client.RestTemplate
import java.io.IOException
import java.util
import java.util.Map

object HdfsChmodUtils {
  protected lazy implicit val logger = LoggerFactory.getLogger(getClass)
  def getTableLocation(dataBase:String, table:String, user:String):String = {
    val gateWayHost = GATEWAY_URL.getValue
    val url = s"$gateWayHost/api/rest_j/v1/datasource/getTableLocation?database=$dataBase&tableName=$table"
    val restTemplate =  new RestTemplate()
    val headers = new HttpHeaders
    headers.setContentType(MediaType.APPLICATION_JSON)
    headers.add("Token-User", user)
    headers.add("Token-Code", "LINKISCLI-AUTH")

    val entity = new HttpEntity[AnyRef](headers)
    val response = restTemplate.exchange(url, HttpMethod.GET, entity, classOf[util.Map[_, _]]).getBody
    val path = response.get("data").asInstanceOf[Map[String, String]].get("tablePath")
    path
  }

  def chmodDbTablePath(dataBase:String, table:String, user:String) = {
    logger.info(s"chmod dataBase.table path: $dataBase -> $table")
    val dbPath = s"/user/hive/warehouse/$dataBase.db"
    try{
      val isExist = HDFSUtils.getHDFSUserFileSystem(user).exists(new Path(dbPath))
      if (isExist) {
        val owner = HDFSUtils.getHDFSUserFileSystem(user).getFileStatus(new Path(dbPath)).getOwner()
        val permission = HDFSUtils.getHDFSUserFileSystem(owner).getAclStatus(new Path(dbPath)).getPermission
        if(!FsAction.READ_EXECUTE.or(permission.getOtherAction).SYMBOL.equals(permission.getOtherAction.SYMBOL)){
          val newPermission = new FsPermission(permission.getUserAction, permission.getGroupAction, FsAction.READ_EXECUTE.or(permission.getOtherAction))
          HDFSUtils.getHDFSUserFileSystem(owner).setPermission(new Path(dbPath), newPermission)
          logger.info(s"dbPath: $owner chmod $dbPath")
        }
      }
    }catch {
      case e: Exception => {
        logger.error(e.getMessage())
      }
    }
    try {
      val tablePath = getTableLocation(dataBase, table, user)
      val pathStr = new Path(tablePath).toUri.getPath
      var dirPath = ""
      pathStr.split("/").toList.filter(p => !p.isEmpty).foreach(
        item => {
          dirPath += s"/$item"
          setReadExecute(dirPath, user)
        }
      )
      val isExist = HDFSUtils.getHDFSUserFileSystem(user).exists(new Path(tablePath))
      if (isExist) {
        setPermissionIter(tablePath, user)
        logger.info(s"tablePath: $user chmod $tablePath")
      }
    }catch {
      case e: Exception => {
        logger.error(e.getMessage())
      }
    }
  }

  def setReadExecute(dest: String, user: String):Boolean = {
    logger.info(s"try chmod subPath: $user chmod $dest")
    val fs = HDFSUtils.getHDFSUserFileSystem(user)
    val filePath = new Path(dest)
    val isExist = fs.exists(filePath)
    if(isExist){
      val owner = fs.getFileStatus(filePath).getOwner()
      val ownerFs = HDFSUtils.getHDFSRootUserFileSystem(owner)
      val permission = ownerFs.getAclStatus(filePath).getPermission
      if(!FsAction.READ_EXECUTE.or(permission.getOtherAction).SYMBOL.equals(permission.getOtherAction.SYMBOL)){
        val newPermission = new FsPermission(permission.getUserAction, permission.getGroupAction, FsAction.READ_EXECUTE.or(permission.getOtherAction))
        ownerFs.setPermission(filePath, newPermission)
        logger.info(s"tableSubPath: $owner chmod $dest")
      }
    }
    true
  }

  @throws[IOException]
  def setPermissionIter(dest: String, user: String): Boolean = {
    val fs = HDFSUtils.getHDFSUserFileSystem(user)
    val filePath = new Path(dest)

    val owner = fs.getFileStatus(filePath).getOwner()
    val isDir = fs.getFileStatus(filePath).isDirectory()

    val ownerFs = HDFSUtils.getHDFSRootUserFileSystem(owner)
    val permission = ownerFs.getAclStatus(filePath).getPermission
    if(!FsAction.READ_EXECUTE.or(permission.getOtherAction).SYMBOL.equals(permission.getOtherAction.SYMBOL)){
      val newPermission = new FsPermission(permission.getUserAction, permission.getGroupAction, FsAction.READ_EXECUTE.or(permission.getOtherAction))
      ownerFs.setPermission(filePath, newPermission)
      logger.info(s"tableSubPath: $owner chmod $dest")
    }
    if (!isDir) return true
    val status = ownerFs.listStatus(filePath) //获取文件目录
    for (file <- status) {
      setPermissionIter(file.getPath.toUri.getPath, user)
    }
    true
  }

}
